package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object StudentScoresComplex {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Student Scores Complex").setMaster("local[1]")
    val sc = new SparkContext(conf)

    // 定义更复杂的学生数据：(姓名, 班级, 电话, 地址, (数学, 英语, 物理, 化学, 生物))
    val studentData = Seq(
      ("Alice", "Class1", "123-4567", "123 Pine St", (95, 88, 92, 85, 90)),
      ("Bob", "Class2", "234-5678", "456 Oak Ave", (78, 85, 76, 82, 79)),
      ("Charlie", "Class1", "345-6789", "789 Maple Dr", (92, 95, 88, 90, 93)),
      ("David", "Class2", "456-7890", "321 Elm St", (85, 82, 88, 84, 86)),
      ("Eva", "Class1", "567-8901", "654 Cedar Ln", (91, 94, 89, 88, 92)),
      ("Frank", "Class3", "678-9012", "987 Birch Rd", (76, 78, 75, 80, 77)),
      ("Grace", "Class3", "789-0123", "147 Walnut Ct", (94, 92, 90, 91, 89)),
      ("Henry", "Class2", "890-1234", "258 Spruce Way", (88, 86, 89, 87, 85)),
      ("Ivy", "Class1", "901-2345", "369 Ash Blvd", (93, 90, 92, 94, 91)),
      ("Jack", "Class3", "012-3456", "741 Willow Pl", (82, 84, 80, 83, 81))
    )

    // 将本地集合并行化为RDD
    val studentsRDD = sc.parallelize(studentData, 1)

    // 首先过滤数学成绩大于等于80的学生，然后计算平均成绩
    val studentsWithAvg = studentsRDD
      .filter { case (_, _, _, _, scores) => scores._1 >= 80 } // 只保留数学成绩大于等于80的学生
      .map { case (name, className, phone, address, scores) =>
        val avg = (scores._1 + scores._2 + scores._3 + scores._4 + scores._5) / 5.0
        (name, className, phone, address, scores, avg)
      }

    // 1. 打印符合条件的学生详细信息和平均分
    println("\n=== 数学成绩>=80的学生详细信息和平均分 ===")
    studentsWithAvg.collect().foreach { case (name, className, phone, address, scores, avg) =>
      println(f"姓名: $name%-10s 班级: $className%-8s 平均分: $avg%.2f")
      println(f"电话: $phone%-12s 地址: $address")
      println(f"成绩详情 - 数学: ${scores._1}%-3d 英语: ${scores._2}%-3d 物理: ${scores._3}%-3d 化学: ${scores._4}%-3d 生物: ${scores._5}%-3d")
      println("----------------------------------------")
    }

//    // 2. 计算并打印符合条件学生的班级平均成绩
//    println("\n=== 数学成绩>=80的学生班级平均成绩 ===")
//    val classAvgScores = studentsWithAvg
//      .map { case (_, className, _, _, _, avg) => (className, (avg, 1)) }
//      .reduceByKey { case ((sum1, count1), (sum2, count2)) => (sum1 + sum2, count1 + count2) }
//      .mapValues { case (sum, count) => sum / count }
//      .collect()
//      .sortBy(_._1)
//
//    classAvgScores.foreach { case (className, avgScore) =>
//      println(f"$className: $avgScore%.2f")
//    }


//
//    // 3. 统计符合条件的学生数量
//    val totalStudents = studentsWithAvg.count()
//    println(s"\n符合条件的学生总数: $totalStudents")
//
//    // 4. 计算符合条件学生的总体平均分
//    val overallAvg = studentsWithAvg
//      .map(_._6) // 提取平均分
//      .mean()
//    println(f"符合条件学生的总体平均分: $overallAvg%.2f")

    // 暂停5分钟以便查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
} 